package com.shujia.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Properties;

public class Code05HDFS2Mysql {
    public static void main(String[] args) throws IOException, ClassNotFoundException, SQLException {

        FileSystem fileSystem = null;
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://master:9000");
        conf.set("dfs.replication", "2");
        System.setProperty("HADOOP_USER_NAME", "root");
        try {
            fileSystem = FileSystem.get(conf);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        Path path = new Path("/api/students.txt");
        FSDataInputStream open = fileSystem.open(path);

        // 对IO流进行包装
        //  InputStreamReader 是字节流和字符流的转换流
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(open));

        //1、注册驱动
        Class.forName("com.mysql.jdbc.Driver");

        // 2.创建连接
        Properties properties = new Properties();
        properties.load(new FileReader("hadoopCode/src/main/resources/mysql.properties"));
        String url = properties.getProperty("url");
        String user = properties.getProperty("user");
        String passwd = properties.getProperty("passwd");
        Connection connection = DriverManager.getConnection(url, user, passwd);

        Statement statement = connection.createStatement();

        String read;
        while ((read = bufferedReader.readLine()) != null) {
            String[] columns = read.split(",");
            // 数据处理
            String id = columns[0];
            String name = columns[1];
            int age = Integer.parseInt(columns[2]);
            String gender = columns[3];
            String clazz = columns[4];

            // 写入Mysql
            statement.executeUpdate(
                    String.format("insert into bigdata26.students values('%s','%s',%d,'%s','%s')", id, name, age, gender, clazz)
            );

        }


        // 关闭操作
        bufferedReader.close();
        fileSystem.close();
        statement.close();
        connection.close();

    }
}
